package com.adcc.ags.msgStore.managed;

import com.adcc.ags.msgStore.DBConnectionFactory;
import com.adcc.ags.msgStore.MQConnectionFactory;
import com.adcc.ags.msgStore.entity.db.DBInfo;
import com.adcc.ags.msgStore.entity.mq.MQInfo;
import com.adcc.ags.msgStore.entity.mq.MQState;
import com.adcc.ags.msgStore.pool.C3P0Pool;
import com.adcc.ags.msgStore.entity.db.DBState;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.MQConnectionPool;
import com.adcc.utility.mq.transfer.MQConnectionPoolFactory;
import com.adcc.utility.mq.transfer.MQTransfer;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.adcc.utility.mq.transfer.ibm.IBMMQTransfer;
import io.dropwizard.lifecycle.Managed;
import io.dropwizard.setup.Environment;

import java.util.concurrent.TimeUnit;


/**
 * 消息转发
 */
public class MessageForward implements Managed {

    // 下行线程名称
    public static final String CONSTANT_MESSAGE_FORWARD_THREAD_NAME = "Thread.MessageForward";

    // 单例对象
    private static MessageForward instance;

    // 获取报文线程
    private MessageForwardRunnable messageForwardRunnable;

    // MQTransfer
    private MQTransfer transfer = new IBMMQTransfer();

    // MQConfiguration
    MQConfiguration configuration;

    // MQConnectionPool
    MQConnectionPool pool;

    private MQConnectionFactory mqConnectionFactory;

    private DBConnectionFactory dbConnectionFactory;

    private MQInfo mqInfo = MQInfo.getInstance();

    private DBInfo dbInfo = DBInfo.getInstance();

    /**
     * 构造方法
     * */
    private MessageForward(){}

    /**
     * 单例方法
     * */
    public static MessageForward getInstance() {
        if (instance == null) {
            instance = new MessageForward();
        }
        return instance;
    }

    /**
     * 获取MQ信息
     * @return
     */
    public MQInfo getMQInfo() {
        return mqInfo;
    }

    /**
     * 获取数据库信息
     * @return
     */
    public DBInfo getDBInfo() {
        return dbInfo;
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment,MQConnectionFactory mqConnectionFactory,DBConnectionFactory dbConnectionFactory) throws Exception{
        if (environment == null) {
            new NullPointerException("environment is null");
        }
        if (mqConnectionFactory == null) {
            throw new NullPointerException("mqConnectionFactory is null");
        }
        if (dbConnectionFactory == null) {
            throw new NullPointerException("dbConnectionFactory is null");
        }
        this.mqConnectionFactory = mqConnectionFactory;
        this.dbConnectionFactory = dbConnectionFactory;
        // 初始化mqInfo
        mqInfo.setHost(mqConnectionFactory.getIp());
        mqInfo.setPort(mqConnectionFactory.getPort());
        mqInfo.setQm(mqConnectionFactory.getQueueManager());
        mqInfo.setChannel(mqConnectionFactory.getRecChannelName());
        // 设置MQ连接参数
        configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(mqConnectionFactory.getIp(), mqConnectionFactory.getPort(),
                mqConnectionFactory.getQueueManager(), mqConnectionFactory.getRecChannelName());
        transfer.setConfiguration(configuration);
        // 设置MQ连接池参数
        pool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
        ((IBMMQConnectionPool)pool).setActiveMode(mqConnectionFactory.getActiveMode());
        ((IBMMQConnectionPool)pool).setTimeout(mqConnectionFactory.getTimeout());
        ((IBMMQConnectionPool)pool).setMaxConnections(mqConnectionFactory.getMaxConnections());
        ((IBMMQConnectionPool)pool).setMaxIdelConnections(mqConnectionFactory.getMaxIdelConnections());
        transfer.setConnectionPool(pool);
        // 初始化dbInfo
        dbInfo.setUrl(dbConnectionFactory.getUrl());
        dbInfo.setUsername(dbConnectionFactory.getUsername());
        // 设置DB连接池参数
        C3P0Pool.setDbConnectionFactory(this.dbConnectionFactory);
        pool.init(configuration);
        environment.lifecycle().manage(this);
    }

    /**
     * 连接MQ
     */
    private boolean connectMQ() {
        try {
            if (transfer.isConnected()) {
                Log.info(MessageForward.class.getName(), "MessageForward connected mq host:" + mqConnectionFactory.getIp() + " port:" + mqConnectionFactory.getPort() + " channel:" + mqConnectionFactory.getRecChannelName() + " qm:" + mqConnectionFactory.getQueueManager());
                mqInfo.setState(MQState.CONNECTED);
                return true;
            } else {
                Log.error(MessageForward.class.getName(), "connect to ibm mq failed!");
                return false;
            }
        } catch (Exception ex) {
            Log.error(MessageForward.class.getName(), "connectMQ() error", ex);
            return false;
        }
    }

    /**
     * 连接数据库
     * @return
     */
    private boolean connectDB(){
        try {
            if (C3P0Pool.checkConnection()) {
                Log.info(MessageForward.class.getName(), "MessageForward connected db url:" + dbConnectionFactory.getUrl() + " username:" + dbConnectionFactory.getUsername());
                dbInfo.setState(DBState.CONNECTED);
                return true;
            } else {
                Log.error(MessageForward.class.getName(), "connect to database failed!");
                return false;
            }
        } catch (Exception ex) {
            Log.error(MessageForward.class.getName(), "connectDB() error", ex);
            return false;
        }
    }

    @Override
    public void start() throws Exception {
        try {
            C3P0Pool.init();
            messageForwardRunnable = new MessageForwardRunnable();
            new Thread(messageForwardRunnable,CONSTANT_MESSAGE_FORWARD_THREAD_NAME).start();
        } catch (Exception ex) {
            Log.error(MessageForward.class.getName(), "start() error", ex);
        }
    }

    @Override
    public void stop() throws Exception {
        try {
            if (messageForwardRunnable != null) {
                messageForwardRunnable.close();
            }
            if(pool!=null){
                pool.dispose();
            }
            mqInfo.setState(MQState.DISCONNECTED);
            dbInfo.setState(DBState.DISCONNECTED);
        } catch (Exception ex) {
            Log.error(MessageForward.class.getName(), "stop() error", ex);
        }
    }

    class MessageForwardRunnable implements Runnable{

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                isStarted = true;
                while (isStarted) {
                    try {
                        // 如果MQ和DB连接状态正常则获取并保存报文
                        if (mqInfo.getState() == MQState.CONNECTED && dbInfo.getState() == DBState.CONNECTED) {
                            try {
                                Message message = transfer.receiveQueue(mqConnectionFactory.getRecQueueName());
                                if(message!=null){
                                    MsgProxy.getInstance().getMsgStoreActor().tell(message,MsgProxy.getInstance().getMsgStoreActor());
                                    Log.info(MessageForward.class.getName(), "receive message from queue:" + mqConnectionFactory.getRecQueueName() + " message:\r\n" + message);
                                }
                            } catch (Exception ex) {
                                Log.error(MessageForward.class.getName(), "receive message from AGS error", ex);
                                mqInfo.setState(MQState.CONNECTING);
                            }
                        } else {
                            if (mqInfo.getState() != MQState.CONNECTED) {
                                if (!connectMQ()) {
                                    mqInfo.setState(MQState.CONNECTING);
                                }
                            }

                            if (dbInfo.getState() != DBState.CONNECTED ) {
                                if (!connectDB()) {
                                    dbInfo.setState(DBState.CONNECTING);
                                }
                            }
                            TimeUnit.SECONDS.sleep(1);
                        }
                    } catch (Exception ex) {
                        Log.error(MessageForwardRunnable.class.getName(), "process msg failed", ex);
                        mqInfo.setState(MQState.CONNECTING);
                    }
                }
            } catch (Exception ex) {
                Log.error(MessageForwardRunnable.class.getName(), "run() error", ex);
            }
        }
    }
}
